package com.zynn.service.module.canal.scheduling;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.Message;
import com.zynn.common.core.utils.SpringContextUtil;
import com.zynn.service.module.canal.constant.DatabaseNameConstant;
import com.zynn.service.module.canal.constant.TableNameConstant;
import com.zynn.service.module.canal.event.DeleteCanalEvent;
import com.zynn.service.module.canal.event.InsertCanalEvent;
import com.zynn.service.module.canal.event.UpdateCanalEvent;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Arrays;
import java.util.List;


/**
 * @author jerry-liqi
 *
 * 计划任务读取数据库日志
 *
 * Canal配置
 */
@Component
@Slf4j
public class CanalScheduling {


    /** 监听表的集合 */
    public static List<String> listenTables = Arrays.asList(
            TableNameConstant.GROUPS,
            TableNameConstant.USERS,
            TableNameConstant.USER_FRIEND,
            TableNameConstant.LIKE,
            TableNameConstant.COMMENTS,
            TableNameConstant.REMINDUSER,
            TableNameConstant.USER_SHARE_EVENT_RECORD,

            TableNameConstant.GROUPMEMBERS,
            TableNameConstant.PICTURES,

            TableNameConstant.BASE_USER_INFO,

            TableNameConstant.SCHOOL_BE_INVITED_RECORD,
            TableNameConstant.SCHOOL_INVITE_RECORD,
            TableNameConstant.SCHOOL_POPULARITY_INVITER,
            TableNameConstant.SCHOOL_POPULARITY_LIKE,
            TableNameConstant.SCHOOL_SUBJECT_SHOW_PK,
            TableNameConstant.EVENTS,
            TableNameConstant.SCHOOL_POPULARITY_STAR_LIGHT,
            TableNameConstant.USER_ATTENTION_RELATION
            );

    @Resource
    private CanalConnector canalConnector;

    @Scheduled(fixedDelay = 100)
    public void run() {
        try {
            int batchSize = 1000;
            Message message = canalConnector.getWithoutAck(batchSize);
            long batchId = message.getId();
            log.debug("scheduled_batchId=" + batchId);
            try {
                List<Entry> entries = message.getEntries();
                if (batchId != -1 && entries.size() > 0) {
                    entries.forEach(entry -> {
                        if (entry.getEntryType() == EntryType.ROWDATA) {
                            publishCanalEvent(entry);
                        }
                    });
                }
                canalConnector.ack(batchId);
            } catch (Exception e) {
                log.error("发送监听事件失败！batchId回滚,batchId=" + batchId, e);
                canalConnector.rollback(batchId);
            }
        } catch (Exception e) {
            log.error("canal_scheduled异常！", e);
        }
    }

    private void publishCanalEvent(Entry entry) {
        ///log.info("entry："+entry);
        EventType eventType = entry.getHeader().getEventType();
        String table = entry.getHeader().getTableName();
        String database = entry.getHeader().getSchemaName();
        if(!DatabaseNameConstant.yinian.equals(database) && !DatabaseNameConstant.yinian_school.equals(database)) {
            return;
        }
        if(!listenTables.contains(table)) {
            return;
        }
        switch (eventType) {
            case INSERT:
                SpringContextUtil.getApplicationContext().publishEvent(new InsertCanalEvent(entry));
                break;
            case UPDATE:
                SpringContextUtil.getApplicationContext().publishEvent(new UpdateCanalEvent(entry));
                break;
            case DELETE:
                SpringContextUtil.getApplicationContext().publishEvent(new DeleteCanalEvent(entry));
                break;
            default:
                break;
        }
    }

}
